/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.CellFilter;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.client.RowFilter;
import com.nhncorp.neptune.common.TimeLog;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.RowColumnValues;
import com.nhncorp.neptune.tabletserver.DiskSSTable.ColumnValueEntry;

public class RecordSearcher {
  public static final Log LOG = LogFactory.getLog(RecordSearcher.class.getName());

//  private static final RowKey EMPTY_ROWKEY = new RowKey();
  
  private Tablet tablet;
  private int numOfVersion;
  
  public RecordSearcher(Tablet tablet) {
    //this.searchers = searchers;
    this.tablet = tablet;
    this.numOfVersion = tablet.getTable().getNumOfVersion();
  }

  private List<Searchable> initSearchsers(Row.Key rowKey, String columnName, Cell.Key cellKey) throws IOException {
    //반드시 MemorySSTable부터 add되어야 한다.
    //그렇지 않으면 MinorCompaction 도중에 get하는 경우 data가 있는데 없다고 return하는 경우가 발생한다.
    List<Searchable> searchers = new ArrayList<Searchable>();
    searchers.add(tablet.memorySSTable.getSearcher(rowKey, columnName, cellKey));
    searchers.addAll(tablet.diskSSTable.getSearchers(rowKey, columnName, cellKey)); 
    return searchers;
  }
  
  public ColumnValue search(Row.Key rowKey, String columnName, Cell.Key columnKey) throws IOException {
    try {
      TimeLog.checkPoint();
      CellFilter cellFilter = new CellFilter();
      cellFilter.setColumn(columnName, columnKey);
      
      ColumnValue[] result = searchColumnValues(rowKey, cellFilter);
      if(result != null && result.length > 0) {
        ColumnValue columnValue = result[0];
        return columnValue;
      } else {
        return null;
      }
    } finally {
      TimeLog.endCheckPoint("RecordSearcher:search1:" + rowKey);
    }
  }
  
  public ColumnValue[] search(Row.Key rowKey, String columnName) throws IOException {
    try {
      TimeLog.checkPoint();
      CellFilter cellFilter = new CellFilter();
      cellFilter.setColumn(columnName, null);
      
      ColumnValue[] columnValues = searchColumnValues(rowKey, cellFilter);
      
      return columnValues;
      
    } finally {
      TimeLog.endCheckPoint("RecordSearcher:search3:" + rowKey);
    }
  }
  
  /**
   * 여러개의 컬럼에 대한 값을 반환한다.
   * 하나의 Row.Key, 하나의 컬럼에 여러개의 컬럼값이 저장되기 때문에 2차원 배열로 반환한다.
   * @param rowKey
   * @param columnNames
   * @return
   * @throws IOException
   */
  public ColumnValue[][] search(Row.Key rowKey, String[] columnNames) throws IOException {
    if(rowKey == null) {
      throw new IOException("rowkey is null");
    }
    //TODO 컬럼별로 RowKey가 다르게 저장되는 경우 merge하는 로직이 추가 되어야 함
    //TODO 속도 확인 필요
    ColumnValue[][] columnValues = new ColumnValue[columnNames.length][];
    
    for(int i = 0; i < columnNames.length; i++) {
      columnValues[i] = search(rowKey, columnNames[i]);
//      ColumnValue[] columnResult = search(rowKey, columnNames[i]);
//      if(columnResult == null) {
//        columnValues[i] = new ColumnValue[]{};
//      } else {
//        columnValues[i] = columnResult;
//      }
    }
    return columnValues;
  }

  public RowColumnValues[] search(RowFilter rowFilter) throws IOException {
    if(rowFilter.getRowKey() == null || rowFilter.getRowKey().isEmpty()) {
      throw new IOException("rowkey is null");
    }
    
    TimeLog.checkPoint();
    Row.Key rowKey = rowFilter.getRowKey();
    
    List<CellFilter> cellFilters = rowFilter.getCellFilters();
    RowColumnValues[] result = new RowColumnValues[cellFilters.size()];
    
    int index = 0;
    boolean hasData = false;
    for(CellFilter eachCellFilter: cellFilters) {
      RowColumnValues rowColumnValues = new RowColumnValues();
      rowColumnValues.setRowKey(rowKey);
      rowColumnValues.setColumnName(eachCellFilter.getColumnName());
      
      ColumnValue[] columnValues = searchColumnValues(rowKey, eachCellFilter);
      
      if(columnValues != null &&columnValues.length > 0) {
        hasData = true; 
      }
      rowColumnValues.setColumnRecords(columnValues);
      result[index++] = rowColumnValues;
    }
    
    TimeLog.endCheckPoint("RecordSearcher:search5:" + rowFilter.getRowKey());
    if(!hasData) {
      return null;
    } else {
      return result;
    }
  }
  
  private ColumnValue[] searchColumnValues(Row.Key rowKey, CellFilter cellFilter) throws IOException {
    if(rowKey == null) {
      throw new IOException("rowkey is null");
    }
    
    List<Searchable> searchers = initSearchsers(rowKey, 
                                                cellFilter.getColumnName(), 
                                                cellFilter.getCellKey());
    
    try {
      if(searchers.size() == 0) {
        return null;
      }
        
  //    if(searchers.size() == 1) {
  //      return searchers.get(0).nextAll(searchFilter);
  //    }
          
      List<ColumnValue> columnValues = mergeColumnValueFromSearcher(searchers, cellFilter);
      if(columnValues == null || columnValues.size() == 0) {
        return null;
      }

//      for(ColumnValue eachValue: columnValues) {
//        System.out.println("====" + eachValue);
//      }
      ColumnValue[] result = new ColumnValue[columnValues.size()];
      columnValues.toArray(result);
      return result;
    } finally {
    }
  }
  
  private List<ColumnValue> mergeColumnValueFromSearcher(
      List<Searchable> searcherList, 
      CellFilter cellFilter) throws IOException {
    
    int numOfValues = cellFilter.getNumOfValues();
    
    Cell.Key targetColumnKey = cellFilter.getCellKey();
    
    List<ColumnValueEntry> workPlace = new ArrayList<ColumnValueEntry>(searcherList.size());
    
    Searchable[] searchers = searcherList.toArray(new Searchable[]{});
    
    //초기화
    for(int i = 0; i < searchers.length; i++) {
      if(searchers[i] != null) {
        ColumnValue columnValue = searchers[i].next();

        if(columnValue != null) {
          workPlace.add(new ColumnValueEntry(columnValue, i));
        } else {
          searchers[i].close();
          searchers[i] = null;
        }
      }
    }
    
    List<ColumnValue> result = new ArrayList<ColumnValue>();
    
    ValueCollection currentColumnValues = new ValueCollection();
    boolean end = false;
    
    Cell.Key previousColumnKey = null;
    //Compaction 도중 kill 된 경우 서로 다른 MapFile내에 중복된 데이터가 존재할 수 있다.
    //따라서 이전 데이터와 timestamp까지 동일한 경우 처리하지 않는다.
    ColumnValue previousColumnValue = null;    
    
    while(!end) {
      int size = workPlace.size();
      if(size == 0) {
        break;
      }
      
      Collections.sort(workPlace);
      
      ColumnValueEntry winner = workPlace.remove(0);
      
      if(targetColumnKey != null && 
          winner.columnValue.getCellKey() != null &&
          winner.columnValue.getCellKey().compareTo(targetColumnKey) > 0) {
        break;
      }
      
      if(previousColumnKey == null) {
        previousColumnKey = winner.columnValue.getCellKey();
      }
      
      if(!previousColumnKey.equals(winner.columnValue.getCellKey())) {
        //여기서 검색 조건에 맞는 columnValue를 찾아서 result에 add
        List<ColumnValue> machedColumnValues = currentColumnValues.get(cellFilter);
        if(machedColumnValues != null) {
          for(ColumnValue eachColumnValue: machedColumnValues) {
            eachColumnValue.setNumOfValues(currentColumnValues.getValueSize());
            //FIXME 성능관련해서 결과에서 rowKey를 돌려주는 것이 맞는가?
            //machedColumnValue.setRowKey(EMPTY_ROWKEY);
            result.add(eachColumnValue);
          }
        }
        currentColumnValues = new ValueCollection();
      } 
      
      //값이 동일한 경우에는 무시한다.
      if(previousColumnValue == null || !previousColumnValue.equals(winner.columnValue)) {
        currentColumnValues.add(winner.columnValue, numOfVersion);
        previousColumnKey = winner.columnValue.getCellKey();
        previousColumnValue = winner.columnValue;
      }
      
      ColumnValue columnValue = null;
      if(searchers[winner.index] != null) {
        columnValue = searchers[winner.index].next();
      }
      
      if(columnValue != null) {
        workPlace.add(new ColumnValueEntry(columnValue, winner.index));
      } else {
        if(searchers[winner.index] != null) {
          searchers[winner.index].close();
        }
        searchers[winner.index] = null;
        
        end = workPlace.isEmpty();
        for(int i = 0; i < searchers.length; i++) {
          if(searchers[i] != null) {
            columnValue = searchers[i].next();
            if(columnValue != null) {
              workPlace.add(new ColumnValueEntry(columnValue, i));
              end = false;
              break;
            } else {
              searchers[i].close();
              searchers[i] = null;
            }
          }
        }
      }
      
      if(numOfValues > 0 && result.size() >= numOfValues) {
        break;
      }
    } //end of while
    
    if(numOfValues == 0 || result.size() < numOfValues) {
      if(currentColumnValues.getValueSize() > 0) {
        List<ColumnValue> machedColumnValues = currentColumnValues.get(cellFilter);
        if(machedColumnValues != null) {
          for(ColumnValue eachColumnValue: machedColumnValues) {
            eachColumnValue.setNumOfValues(currentColumnValues.getValueSize());
  //        FIXME 성능관련해서 결과에서 rowKey를 돌려주는 것이 맞는가?
            //eachColumnValue.setRowKey(EMPTY_ROWKEY);      
            result.add(eachColumnValue);
          }
        }
      }
    }
    
    //scnner close
    for(int i = 0; i < searchers.length; i++) {
      if(searchers[i] != null) {
        searchers[i].close();
        searchers[i] = null;
      }
    }
    return result;
  }

  public boolean hasValue(String columnName, Row.Key rowKey) throws IOException {
    //List<Searchable> searchers = initSearchsers(rowKey, columnName, null);
    if(tablet.memorySSTable.hasValue(rowKey, columnName)) {
      return true;
    }
    return tablet.diskSSTable.hasValue(rowKey, columnName);
  }
}
